package cn.linjianhui.demo.flink.example;

import cn.linjianhui.demo.flink.example.Function.TopNProcessFunction;
import cn.linjianhui.demo.flink.example.model.ItemViewCount;
import cn.linjianhui.demo.flink.example.model.UserBehavior;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import javax.annotation.Nullable;

public class HotItems {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);
        String filePath = Thread.currentThread().getContextClassLoader().getResource("UserBehavior.csv").getPath();
        DataStream<String> result = env.readTextFile(filePath).map(new MapFunction<String, UserBehavior>() {
            @Override
            public UserBehavior map(String line) throws Exception {
                String[] param = line.split(",");
                return new UserBehavior(Long.parseLong(param[0]), Long.parseLong(param[1]),
                        Integer.parseInt(param[2]), param[3], Long.parseLong(param[4]) * 1000L);
            }
        }).filter((FilterFunction<UserBehavior>) behavior -> behavior.getBehavior().equals("pv"))
                .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<UserBehavior>(){
                    Long currentMaxTimestamp = 0L;

                    @Override
                    public long extractTimestamp(UserBehavior element, long previousElementTimestamp) {
                        long timestamp = element.getTimestamp();
                        currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
                        return timestamp;
                    }

                    @Override
                    public Watermark getCurrentWatermark() {
                        return new Watermark(currentMaxTimestamp);
                    }
                })
                .keyBy("itemId")
                .timeWindow(Time.minutes(60), Time.minutes(5))
//                .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(1)))
                .aggregate(new AggregateFunction<UserBehavior, Long, Long>() {
                    @Override
                    public Long createAccumulator() {
                        return 0L;
                    }

                    @Override
                    public Long add(UserBehavior behavior, Long acc) {
                        return acc + 1;
                    }

                    @Override
                    public Long getResult(Long acc) {
                        return acc;
                    }

                    @Override
                    public Long merge(Long acc1, Long acc2) {
                        return acc1 + acc2;
                    }
                }, new ProcessWindowFunction<Long, ItemViewCount, Tuple, TimeWindow>() {
                    @Override
                    public void process(Tuple key, Context context, Iterable<Long> elements, Collector<ItemViewCount> out) throws Exception {
                        Long itemId = key.getField(0);
                        out.collect(new ItemViewCount(itemId, context.window().getEnd(), elements.iterator().next()));
                    }
                }).keyBy("windowEnd")
                .process(new TopNProcessFunction(3));

        result.print();

        env.execute("TopN HotItems");
    }
}
